package cn.tecnova.test

import cn.tecnova.cd.Base
import cn.tecnova.utils.ConfigHandler
import com.alibaba.fastjson.{JSON, JSONObject}
import com.google.gson.Gson
import org.apache.commons.pool2.impl.GenericObjectPoolConfig
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SQLContext, SaveMode}
import org.elasticsearch.spark.rdd.EsSpark
import redis.clients.jedis.JedisPool

/**
  * description:
  * Rabcheng
  * Date:2019/6/20 15:26
  **/
object Base2Redis {


  def main(args: Array[String]): Unit = {


    val conf = new SparkConf()
      .setAppName("estest")
    //      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .registerKryoClasses(Array(classOf[Base]))
      .set("es.nodes", "172.17.22.16")
      .set("es.port", "9600")
    val sc = new SparkContext(conf)

    val sQLContext = new SQLContext(sc)


    val rdd: RDD[(String, String)] = EsSpark.esJsonRDD(sc, "base/zhangguoao", "{\"query\":{\"bool\":{\"must\":[{\"term\":{\"cityCode\":\"320600\"}}],\"must_not\":[],\"should\":[]}},\"from\":0,\"size\":10,\"sort\":[],\"aggs\":{}}")

    rdd.foreachPartition(iter=>{

      val productor: KafkaProducer[String, String] = new KafkaProducer[String, String](ConfigHandler.kafkaProps)
      iter.foreach(tp=>{
        productor.send(new ProducerRecord[String, String]("CDBase", tp._2))
      })

    })

    /*rdd.foreachPartition(iter => {
//      val pool = new JedisPool("172.17.22.18", 8014)
      val pool = new JedisPool(new GenericObjectPoolConfig, "172.17.22.18", 8014, 1000, "123456", 0)
      val jedis = pool.getResource
      iter.foreach(tp => {
        println(tp._2)
        jedis.rpush("base", tp._2)
        println("写入成功...")
      })

    })*/

    import sQLContext.implicits._

    val res: RDD[Base] = rdd.map(tp => {

      val jsonObj: JSONObject = JSON.parseObject(tp._2)
      Base(
      jsonObj.getInteger("id"),
      jsonObj.getString("keyNo"),
      jsonObj.getString("name"),
      jsonObj.getString("No"),
      jsonObj.getString("belongOrg"),
      jsonObj.getString("operName"),
      jsonObj.getString("startDate"),
      jsonObj.getString("endDate"),
      jsonObj.getString("status"),
      jsonObj.getString("updatedDate"),
      jsonObj.getString("shortStatus"),
      jsonObj.getString("creditCode"),
      jsonObj.getString("registCapi"),
      jsonObj.getString("econKind"),
      jsonObj.getString("industryCode"),
      jsonObj.getString("subIndustryCode"),
      jsonObj.getString("industry"),
      jsonObj.getString("subIndustry"),
      jsonObj.getString("scope"),
      jsonObj.getString("contactNumber"),
      jsonObj.getString("email"),
      jsonObj.getString("webSite"),
      jsonObj.getString("imageUrl"),
      jsonObj.getString("orgNo"),
      jsonObj.getString("englishName"),
      jsonObj.getString("product"),
      jsonObj.getString("type"),
      jsonObj.getString("originalName"),
      jsonObj.getString("hitField"),
      jsonObj.getString("hitValue"),
      jsonObj.getString("address"),
      jsonObj.getString("provinceCode"),
      jsonObj.getString("cityCode"),
      jsonObj.getString("countyCode"),
      jsonObj.getString("province"),
      jsonObj.getString("city"),
      jsonObj.getString("county"),
      jsonObj.getString("registCapiNum")
      )
    })
    res.toDF().write.mode(SaveMode.Append).jdbc("jdbc:mysql://www.slave4.com:3306/tec_base2", "nantong", ConfigHandler.props2)


    //    println("南通市数据条数为："+rdd.count())


    sc.stop()


  }

}


case class Base(

                 var id: Int,
                 var keyNo: String,
                 var name: String,
                 var no: String,
                 var belongOrg: String,
                 var operName: String,
                 var startDate: String,
                 var endDate: String,
                 var status: String,
                 var updatedDate: String,
                 var shortStatus: String,
                 var creditCode: String,
                 var registCapi: String,
                 var econKind: String,
                 var industryCode: String,
                 var subIndustryCode: String,
                 var industry: String,
                 var subIndustry: String,
                 var scope: String,
                 var contactNumber: String,
                 var email: String,
                 var webSite: String,
                 var imageUrl: String,
                 var orgNo: String,
                 var englishName: String,
                 var product: String,
                 var Type: String,
                 var originalName: String,
                 var hitField: String,
                 var hitValue: String,
                 var address: String,
                 var provinceCode: String,
                 var cityCode: String,
                 var countyCode: String,
                 var province: String,
                 var city: String,
                 var county: String,
                 var registCapiNum: String

               )
